package com.aotain.metis.bolt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import com.aotain.common.CommonFunction;
import com.aotain.apollo.utils.*;
import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.storm.AbstractBolt;

/**
 * 
 * 源IP访问域名统计
 * <功能详细描述>
 * 
 * @author  Turk
 * @version  [版本号, 2017年3月16日]
 * @see  [相关类/方法]
 * @since  [产品/模块版本]
 */
public class DomainStatBolt extends AbstractBolt{

	/**
	 * serialVersionUID
	 */
	private static final long serialVersionUID = 1L;
	
	private HashMap<String,Long> mapDomainStat = new HashMap<String,Long>();
	private HashMap<String,Long> mapDomainIPStat = new HashMap<String,Long>();
	
	@Override
	public void execute(Tuple input) {
		// TODO Auto-generated method stub
				try {
					if(isTickTuple(input)) {
						//定时入库
						emitCountingData(collector);
						collector.ack(input);
					} else {
						//内存记录IP更新情况
						countInLocal(input); 
						collector.ack(input);
					}
				} catch (Exception e) {			
					e.printStackTrace();
					collector.fail(input);
					throw new FailedException("StatBoltMin throws an exception!",e);
				}
	}

	
	private void countInLocal(Tuple tuple) {
		try
		{
			String SourceIP = tuple.getStringByField("sip").trim();
			String domainname = tuple.getStringByField("domainname").trim();
			//String url = tuple.getStringByField("url").trim();
			//d1.sina.com.cn/201703/16/1448095.swf
			String refer = tuple.getStringByField("refer").trim();
			
			
			
			if(domainname.isEmpty() || SourceIP.isEmpty())
				return;
			
			if(SourceIP.equals("192.168.3.198"))
				return;
			
			if(!refer.isEmpty()) {
				domainname = refer.split("/",-1)[0];
			}
			
			//判断是否属于局域网IP段范围内
			domainname = CommonFunction.getTopDomain(domainname);
			if(domainname.isEmpty())
				return;

			
			long nDomainNum = 0L;
			long nDomainIPNum = 0L;
			
			if(mapDomainStat.containsKey(domainname)) {
				nDomainNum = mapDomainStat.get(domainname) + 1;
			} else {
				nDomainNum = 1;
			}
			mapDomainStat.put(domainname, nDomainNum);
			
			String domainnameIP = String.format("%s_%s", domainname, SourceIP);
			if(mapDomainIPStat.containsKey(domainnameIP)) {
				nDomainIPNum = mapDomainIPStat.get(domainnameIP) + 1;
			} else {
				nDomainIPNum = 1;
			}
			mapDomainIPStat.put(domainnameIP, nDomainIPNum);
			
			
		} catch (Exception ex) {
			Logger.getLogger(DomainStatBolt.class).error("=====countInLocal=========ERROR=================",ex);
		}
		
	}
	
	
	private void emitCountingData(OutputCollector collector) {
		
		try {
			
			HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
			SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
			Date date = new Date();
			String dateStr_d = sdf_d.format(date);
			
			//--------METIS_DOMAIN_ALL---------
			for(Entry<String, Long> domainEntry : mapDomainStat.entrySet()) {
				String domainname = domainEntry.getKey();
				String rowkey = String.format("%s_%s", dateStr_d, domainname);
				long value = domainEntry.getValue();
				
				hbaseInstance.Add("METIS_DOMAIN_ALL", rowkey, "cf", "REPORTTIME", dateStr_d);
				hbaseInstance.Add("METIS_DOMAIN_ALL", rowkey, "cf", "DOMAIN", domainname);
				hbaseInstance.incerment("METIS_DOMAIN_ALL", rowkey, "cf:HTTPNUM", value);
			}
			mapDomainStat.clear();
			
			//--------METIS_DOMAIN_IP---------
			for(Entry<String, Long> domainEntry : mapDomainIPStat.entrySet()) {
				String domainnameIP = domainEntry.getKey();
				String domainname = domainnameIP.split("_",-1)[0];
				String sourceIP = domainnameIP.split("_",-1)[1];
				long value = domainEntry.getValue();
				
				String rowkey1 = String.format("%s_%s_%s", domainname, dateStr_d, sourceIP);
				hbaseInstance.Add("METIS_DOMAIN_IP", rowkey1, "cf", "REPORTTIME", dateStr_d);
				hbaseInstance.Add("METIS_DOMAIN_IP", rowkey1, "cf", "DOMAIN", domainname);
				hbaseInstance.Add("METIS_DOMAIN_IP", rowkey1, "cf", "IP", sourceIP);
				hbaseInstance.incerment("METIS_DOMAIN_IP", rowkey1, "cf:HTTPNUM", value);
				
				
				String rowkey2 = String.format("%s_%s_%s", sourceIP, dateStr_d, domainname);
				hbaseInstance.Add("METIS_DOMAIN_IP", rowkey2, "cf", "REPORTTIME", dateStr_d);
				hbaseInstance.Add("METIS_DOMAIN_IP", rowkey2, "cf", "DOMAIN", domainname);
				hbaseInstance.Add("METIS_DOMAIN_IP", rowkey2, "cf", "IP", sourceIP);
				hbaseInstance.incerment("METIS_DOMAIN_IP", rowkey2, "cf:HTTPNUM", value);
			}
			mapDomainIPStat.clear();
			
			
		} catch (Exception ex) {
			Logger.getLogger(DomainStatBolt.class).error("=====emitCountingData=========ERROR=================",ex);
		}
		
	}
	
	
	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,300); //5分钟统计一次
		return conf;
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		
	}

}
